﻿#region LicenseHeader

// Copyright 2012 The Trustees of Indiana University.  All rights reserved.
// 
// The Indiana University licenses this file to you under the Apache License, 
// Version 2.0 (the "License"); you may not use this file except in compliance 
// with the License.  You may obtain a copy of the License at 
// 
// http://www.apache.org/licenses/LICENSE-2.0
// 
// Unless required by applicable law or agreed to in writing, software 
// distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
//  WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.  
// See the License for the specific language governing permissions and 
// limitations under the License.
// 
// @author Thilina Gunarathne (tgunarat@indiana.edu)

#endregion

using System;
using System.Collections.Generic;
using System.Diagnostics;
using System.Linq;
using System.ServiceModel;
using System.Threading.Tasks;
using Microsoft.WindowsAzure.ServiceRuntime;

namespace AzureMRCore.DataCommunication
{
    public class DataTransferClient
    {
        internal static void NotifyAllNodes(DataItem item, ChannelFactory<IDataXferService> factory)
        {
            RoleInstance current = RoleEnvironment.CurrentRoleInstance;
            item.WorkerID = current.Id;


            // iterate over all instances of the internal endpoint except the current role - no need to notify itself
            IEnumerable<RoleInstanceEndpoint> endPoints = current.Role.Instances
                .Select(instance => instance.InstanceEndpoints["NotificationService"]);
            Parallel.ForEach(endPoints, ep =>
                                            {
                                                var address =
                                                    new EndpointAddress(
                                                        String.Format("net.tcp://{0}/NotificationService", ep.IPEndpoint));
                                                IDataXferService client = factory.CreateChannel(address);

                                                try
                                                {
                                                    client.Broadcast(item);
                                                    ((ICommunicationObject) client).Close();
                                                }
                                                catch (TimeoutException timeoutException)
                                                {
                                                    Trace.TraceError(
                                                        "Unable to notify worker role instance '{0}'. The service operation timed out. {1}",
                                                        ep.RoleInstance.Id, timeoutException.Message);
                                                    ((ICommunicationObject) client).Abort();
                                                }
                                                catch (CommunicationException communicationException)
                                                {
                                                    Trace.TraceError(
                                                        "Unable to notify worker role instance '{0}'. There was a communication problem. {1} - {2}",
                                                        ep.RoleInstance.Id, communicationException.Message,
                                                        communicationException.StackTrace);
                                                    ((ICommunicationObject) client).Abort();
                                                }
                                            });
        }

        internal static void TreeBroadcast(TreeBCastItem item, int min, int max,
                                           ChannelFactory<IDataXferService> factory, int maxParallel)
        {
            RoleInstance current = RoleEnvironment.CurrentRoleInstance;
            item.WorkerID = current.Id;

            var nodesList = new List<int>();
            double count = (max - min) + 1; // 31
            var increment = (int) Math.Ceiling(count/maxParallel); //4

            string temp = "";
            for (int i = min; i <= max; i += increment) // 1,5,9,13,17,21,25,29
            {
                nodesList.Add(i);
                temp = temp + " " + i;
            }
            item.Recurse = (increment > 1);
            Trace.WriteLine("Sending TreeBCast data." + current.Id + "  min" + min + "  max" + max + "   " + temp);

            Parallel.ForEach(nodesList, nodeIndex =>
                                            {
                                                int minLocal = nodeIndex + 1;
                                                int maxLocal = Math.Min((nodeIndex + increment - 1), max);
                                                RoleInstance roleInstance =
                                                    current.Role.Instances.First(
                                                        instance => instance.Id.EndsWith("_" + nodeIndex));
                                                RoleInstanceEndpoint ep = roleInstance.InstanceEndpoints[
                                                    "NotificationService"];
                                                var address =
                                                    new EndpointAddress(
                                                        String.Format("net.tcp://{0}/NotificationService", ep.IPEndpoint));

                                                int retryCount = 0;
                                                RetryLabel:
                                                IDataXferService client = factory.CreateChannel(address);
                                                try
                                                {
                                                    client.TreeBroadcast(item, minLocal, maxLocal, maxParallel);
                                                    ((ICommunicationObject) client).Close();
                                                }
                                                catch (Exception exception)
                                                {
                                                    ((ICommunicationObject) client).Close();
                                                    if (retryCount < 3)
                                                    {
                                                        retryCount++;
                                                        goto RetryLabel;
                                                    }
                                                    Trace.TraceError(
                                                        "Unable to notify worker role instance '{0}'. The service operation timed out. {1}",
                                                        nodeIndex, exception.Message);
                                                }
                                            });
        }

        internal static void TransferIntermediateData(DataItem item, string workerID,
                                                      ChannelFactory<IDataXferService> factory)
        {
            RoleInstance current = RoleEnvironment.CurrentRoleInstance;
            item.WorkerID = current.Id;

            // iterate over all instances of the internal endpoint except the current role - no need to notify itself
            RoleInstanceEndpoint endPoint = current.Role.Instances.Where(p => p.Id.Equals(workerID))
                .Select(instance => instance.InstanceEndpoints["NotificationService"]).First();

            var address =
                new EndpointAddress(
                    String.Format("net.tcp://{0}/NotificationService", endPoint.IPEndpoint));
            IDataXferService client = factory.CreateChannel(address);

            try
            {
                client.PutReduceData(item);
                ((ICommunicationObject) client).Close();
            }
            catch (TimeoutException timeoutException)
            {
                Trace.TraceError(
                    "Unable to notify worker role instance '{0}'. The service operation timed out. {1}",
                    endPoint.RoleInstance.Id, timeoutException.Message);
                ((ICommunicationObject) client).Abort();
            }
            catch (CommunicationException communicationException)
            {
                Trace.TraceError(
                    "Unable to notify worker role instance '{0}'. There was a communication problem. {1} - {2}",
                    endPoint.RoleInstance.Id, communicationException.Message,
                    communicationException.StackTrace);
                ((ICommunicationObject) client).Abort();
            }
        }
    }
}